package com.wzf.task.sink;

import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.util.Date;

@Slf4j
public class CustomerSink extends RichSinkFunction<String> {

    public volatile static Date time = new Date();

    public volatile static Boolean timerIsStart = false;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        startTimer();

    }

    @Override
    public void invoke(String json, Context context) {
        time = new Date();
        //HandlerChain.handerChain(JSON.parseObject(json));
        log.info("CustomerSink 接收数据变化 {}", json);//解析数据，同步到其他数据库
    }
    private void startTimer(){
        synchronized (timerIsStart){
            if(timerIsStart){
                return;
            }
            //动态的添加定时任务每1分钟执行一次
            CronUtil.schedule("0 */40 * * * ?", new Task() {
                @Override
                public void execute() {
                    long minute = DateUtil.between(new Date(),CustomerSink.time, DateUnit.MINUTE);
                    log.info("距离上次接收到消息已经 {} 分钟",minute);
                    if(minute>=60){
                        log.info("flinkcdc 已经"+(minute/60)+"小时没有监听到数据了，修改scb_cdc_test表数据 测试是否还可以使用 ");
                    }
                }
            });
            //支持秒级
            CronUtil.setMatchSecond(true);
            //开启定时任务
            CronUtil.start(true);
            timerIsStart = true;
        }
    }
}